package com.atguigu;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
//import org.apache.flink.table.descriptors.HBase;
import org.apache.flink.table.descriptors.Schema;
import org.apache.flink.table.descriptors.Rowtime;
import org.apache.flink.table.descriptors.FormatDescriptor;
//import org.apache.flink.table.descriptors.Json;
import org.apache.flink.table.descriptors.ConnectorDescriptor;
import org.apache.flink.table.descriptors.FileSystem;
import org.apache.flink.table.descriptors.FormatDescriptor;
//import org.apache.flink.table.descriptors.CatalogDescriptor;
import org.apache.flink.table.descriptors.ConnectorDescriptor;
import org.apache.flink.table.descriptors.Rowtime;
import org.apache.flink.table.descriptors.Schema;
import org.apache.flink.table.descriptors.StreamTableDescriptor;
//import org.apache.flink.table.descriptors.StreamTableEnvironment;
import org.apache.flink.table.descriptors.StreamTableDescriptor;
import org.apache.flink.table.descriptors.TableDescriptor;
import org.apache.flink.table.descriptors.FileSystem;
import org.apache.flink.table.descriptors.FormatDescriptor;
//import org.apache.flink.table.descriptors.CatalogDescriptor;
import org.apache.flink.table.descriptors.ConnectorDescriptor;
import org.apache.flink.table.descriptors.Rowtime;
import org.apache.flink.table.descriptors.Schema;
import org.apache.flink.table.descriptors.StreamTableDescriptor;
//import org.apache.flink.table.descriptors.StreamTableEnvironment;
import org.apache.flink.table.descriptors.StreamTableDescriptor;
import org.apache.flink.table.descriptors.TableDescriptor;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.table.types.utils.TypeConversions;
import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.TableColumn;
//import org.apache.flink.table.api.TableColumns;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;

import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
public class FlinkTableQueryHBase {
    public static void main(String[] args) throws Exception {
        // set up the execution environment
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);
        // read data from mysql table using Flink CDC
        String mysqlSourceDDL = "CREATE TABLE mysql_source (\n" +
                "  id INT,\n" +
                "  name STRING,\n" +
                "  age INT,\n" +
                "  buy_time TIMESTAMP(3),\n" +
                "  phone STRING,\n" +
                "  pay_time TIMESTAMP(3)\n" +
                ") WITH (\n" +
                "  'connector' = 'mysql-cdc',\n" +
                "  'hostname' = 'localhost',\n" +
                "  'port' = '3306',\n" +
                "  'username' = 'root',\n" +
                "  'password' = 'root',\n" +
                "  'database-name' = 'test',\n" +
                "  'table-name' = 'my_table'\n" +
                ")";
        tableEnv.executeSql(mysqlSourceDDL);
        // sink data into HBase table
        String hbaseSinkDDL = "CREATE TABLE hbase_sink (\n" +
                "  name STRING,\n" +
                "  buy_time TIMESTAMP(3),\n" +
                "  id INT,\n" +
                "  age INT,\n" +
                "  phone STRING,\n" +
                "  pay_time TIMESTAMP(3),\n" +
                "  PRIMARY KEY (name, buy_time) NOT ENFORCED\n" +
                ") WITH (\n" +
                "  'connector' = 'hbase-2.2',\n" +
                "  'table-name' = 'my_hbase_table',\n" +
                "  'zookeeper.quorum' = 'localhost:2181',\n" +
                "  'zookeeper.znode.parent' = '/hbase',\n" +
                "  'write.buffer-flush.max-size' = '2mb',\n" +
                "  'write.buffer-flush.max-rows' = '1000',\n" +
                "  'write.buffer-flush.interval' = '2s',\n" +
                "  'format' = 'json',\n" +
                "  'json.timestamp-format.standard' = 'ISO-8601'\n" +
                ")";
        tableEnv.executeSql(hbaseSinkDDL);
        // insert data into HBase table
        String insertQuery = "INSERT INTO hbase_sink\n" +
                "SELECT name, buy_time, id, age, phone, pay_time\n" +
                "FROM mysql_source";
        tableEnv.executeSql(insertQuery);
    }
}